#include <cassert>
#include <ctime>
#include <iostream>

#ifndef WIN32
#include <signal.h>
#endif

#include "miniutil/os_compatible.h"
#include "miniutil/socket.h"
#include "miniutil/socketcmn.h"
#include "miniutil/buffer_alloc.h"
#include "miniutil/threadpool.h"
#include "miniutil/socketmultiplex.h"

/*
	SocketMulplexService accept socket instance and try to check if it can read/write/exception
	it will not create socket, close socket, and so on
	Invoker should control all these resources. 

	must call reset_socket_event to continue;
*/
namespace miniutil { namespace multinet {

	MonSocketInstance::MonSocketInstance()
	{
		socevent = EVT_NONE;
		soc = NULL;
		closeflag = false;
		waitevt = EVT_NONE;
	}

	MonTcpSocketInstance::MonTcpSocketInstance()
		//:recvbuf(32768*4)
	{
		//psendbuf = NULL;
		//sendlen = 0;
		//lenhavesent = 0;

		recvret = 1;
		sendret = 1;
	}

	void MonTcpSocketInstance::sendbuffer(char* pbuf, int len)
	{
		if ( send_queue.size() > 0 )
		{
			miniutil::data::buffer * last_send_buffer = send_queue.back().get();
			if ( last_send_buffer->free_length() >= (unsigned int)len )
			{
				last_send_buffer->append( pbuf, len );
				return;
			}
		}

		miniutil::data::shared_buffer sbuf = miniutil::data::shared_buffer::safenew( len );
		memcpy( sbuf->data(), pbuf, len );
		sbuf->set_length( len );
		this->send_queue.push_back( sbuf );
	}

	MonUdpSocketInstance::UDPMSG::UDPMSG()
	{
		mlen = 0;
		mpbuf = miniutil::data::pool_buffer_alloc::getdefaultinstance()->retrieve_mem(UDPMSG_SIZE);
		soc = 0;
	}

	MonUdpSocketInstance::UDPMSG::~UDPMSG()
	{
		if(mpbuf != NULL)
		{
			miniutil::data::pool_buffer_alloc::getdefaultinstance()->release_mem(mpbuf);
		}
	}

	//---------------------------------------------------------------

	SocketMulplexService::SocketMulplexService()
	{
#ifdef WIN32
	    WSADATA wsdata;
	    if (WSAStartup(MAKEWORD(1, 1), &wsdata)) return ;
	    //if (WSAStartup(MAKEWORD(2, 0), &wsdata)) return ;
#else
        signal(SIGPIPE,SIG_IGN);
#endif
        stopcmd = false;
		stopstus = true;

		max_dquota = 0x7fffffff;
		max_uquota = 0x7fffffff;

        //M_DEBUG("Max QUota:%d\n", max_uquota);
        mprechktime = 0;
        mrecvbufsize = 32768*4;
	}


    static void* WorkingThread(void* para)
    {
        SocketMulplexService* ps = (SocketMulplexService*)para;
        ps->run();
        return NULL;
    }

	void SocketMulplexService::start_service()
	{
		if(!stopstus)
			return;
#ifdef WIN32
	    {
	    DWORD dwThreadID;
	    HANDLE workerhandle = CreateThread( (LPSECURITY_ATTRIBUTES) NULL, /*security attribute*/
				    0,/* size of thread stack, same as main thread by default*/
				    (LPTHREAD_START_ROUTINE)WorkingThread, /* thread entry function */
				    this, 0, &dwThreadID);
	    }
#else
		bool bret = miniutil::threadpool::RunTask( this );
#endif
		stopstus = false;
	}


	void SocketMulplexService::stop_service()
	{
		if(stopstus)
			return;

		stopcmd = true;
		while(!stopstus)
		{
			miniutil::thread::sleep(10);
		}
	}

	int SocketMulplexService::add_tcp_socket(SOCKET soc, int checkevt)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) != malltcpsockets.end())
		{
			malltcpsockets[soc].waitevt = checkevt;
			return 0;
		}
		//malltcpsockets[soc] = MonTcpSocketInstance();
		malltcpsockets[soc].soc = soc;
		malltcpsockets[soc].waitevt = checkevt;
        malltcpsockets[soc].recvbuf.set_buffer_size(mrecvbufsize);
		return 0;
	}

	int SocketMulplexService::add_udp_socket(SOCKET soc, int checkevt)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malludpsockets.find(soc) != malludpsockets.end())
		{
			malludpsockets[soc].waitevt = checkevt;
			return 0;
		}
		malludpsockets[soc] = MonUdpSocketInstance();
		malludpsockets[soc].soc = soc;
		malludpsockets[soc].waitevt = checkevt;
		return 0;
	}

	int SocketMulplexService::del_socket(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) != malltcpsockets.end())
		{
			malltcpsockets[soc].closeflag =true;
			return 0;
		}
		else if(malludpsockets.find(soc) != malludpsockets.end())
		{
			malludpsockets[soc].closeflag =true;
			return 0;
		}

		return -1;
	}
	


	//-----------------------------------
	bool SocketMulplexService::is_readable(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) != malltcpsockets.end())
		{
			MonTcpSocketInstance& ains = malltcpsockets[soc];
			//if(malltcpsockets[soc].socevent& EVT_READ_READY)
			if(ains.recvbuf.have_content() || ains.recvret <= 0)
				return true;
		}
		else if(malludpsockets.find(soc) != malludpsockets.end())
		{
			MonUdpSocketInstance& ains = malludpsockets[soc];
			//if(malludpsockets[soc].socevent& EVT_READ_READY)
			if(ains.vrecvbuf.size() > 0 )
				return true;
		}
		return false;
	}

	bool SocketMulplexService::is_writeable(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) != malltcpsockets.end())
		{
			//if(malltcpsockets[soc].socevent& EVT_WRITE_READY)
			//if(malltcpsockets[soc].psendbuf == NULL)
            if(malltcpsockets[soc].send_queue.size() < 10)
				return true;
		}
		else if(malludpsockets.find(soc) != malludpsockets.end())
		{
			//if(malludpsockets[soc].socevent& EVT_WRITE_READY)
				return true;
		}
		return false;
	}

	bool SocketMulplexService::is_incoming(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return false;
		if(malltcpsockets[soc].socevent& EVT_INCOMING)
        {
            //before accept, keep this flag
            //malltcpsockets[soc].socevent &= (~EVT_INCOMING);
			return true;
        }
		return false;
	}

	bool SocketMulplexService::is_exception(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return false;
		if(malltcpsockets[soc].socevent& EVT_EXCEPTION)
        {
            malltcpsockets[soc].waitevt &= (~EVT_EXCEPTION);
            malltcpsockets[soc].socevent &= (~EVT_EXCEPTION);
			return true;
        }
		return false;
	}

	bool SocketMulplexService::is_connected(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return false;
		if(malltcpsockets[soc].socevent& EVT_CONNECTED)
        {
            malltcpsockets[soc].waitevt &= (~EVT_CONNECTED);
            malltcpsockets[soc].socevent &= (~EVT_CONNECTED);
			return true;
        }
		return false;
	}

	bool SocketMulplexService::is_closed(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return false;
		//if(malltcpsockets[soc].socevent& EVT_CLOSED)
		if(malltcpsockets[soc].recvret <= 0)
			return true;
		return false;
	}

    int SocketMulplexService::check_tcp_connect(SOCKET soc)
    {
        return add_tcp_socket(soc, EVT_CONNECTED | EVT_EXCEPTION);
    }

    int SocketMulplexService::check_tcp_listen(SOCKET soc)
    {
        return add_tcp_socket(soc, EVT_INCOMING);
    }

    int SocketMulplexService::check_tcp_read(SOCKET soc)
    {
        return add_tcp_socket(soc, EVT_READ_READY);
    }

	//----------------------------------------------------------------------------

	//return  -2 other error
	// -1 socket error
	// > 0 ok
	//notice, must have event EVT_WRITE_READY
	int SocketMulplexService::send(SOCKET soc, const char* buf, int len)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return -2;
		MonTcpSocketInstance &ins = malltcpsockets[soc];
		//if(ins.psendbuf != NULL)
		//	return -2;
		ins.sendbuffer(const_cast<char*>(buf), len);

		////assert(ins.socevent & EVT_WRITE_READY);

		//if(ins.sendret <= 0)  //last time send error, so reject to send again.
		//	return ins.sendret;
		//ins.psendbuf = buf;
		//ins.sendlen = len;
		//ins.lenhavesent = 0;

		////ins.socevent &= (~EVT_WRITE_READY);
		return len;
	}

	//return <=0 socket error
	//notice, must have event EVT_READ_READY
	int SocketMulplexService::recv(SOCKET soc, char* buf, int len)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return -1;

		MonTcpSocketInstance &ins = malltcpsockets[soc];

		if(ins.recvbuf.have_content())
			return ins.recvbuf.read(buf, len);

		if(ins.recvret <= 0)
			return ins.recvret;

        M_DEBUG("%d",ins.recvret)
		assert(0);
		//assert(ins.socevent & EVT_READ_READY);
		//if(ins.lenhavereceived < ins.recvlen)
		//{
		//	int surlen = ins.recvlen - ins.lenhavereceived;
		//	int cplen = surlen>len?len:surlen;
		//	memcpy(buf, ins.recvbuf+ins.lenhavereceived, cplen);
		//	ins.lenhavereceived += cplen;
		//	if(ins.lenhavereceived == ins.recvlen)
		//	{
		//		ins.lenhavereceived = 0;
		//		ins.recvlen = 0;
		//		ins.socevent &= (~EVT_READ_READY);
		//	}
		//	return cplen;
		//}

		////assert(0);
		//return -2;
		return 0;
	}

	SOCKET SocketMulplexService::accept(SOCKET soc, sockaddr* addr, int *addrlen)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return -1;

		MonSocketInstance &ins = malltcpsockets[soc];
#ifdef WIN32
		SOCKET connector = ::accept(soc, addr , addrlen );
#else
		SOCKET connector = ::accept(soc, addr , (socklen_t*)addrlen );
#endif
		ins.socevent &= (~EVT_INCOMING);
		return connector;
	}

	int SocketMulplexService::sendto(SOCKET soc, const char* buf,int len,const struct sockaddr* to)
	{
        if(len > MonUdpSocketInstance::UDPMSG_SIZE)
			return -1;
		miniutil::auto_lock kk(&msockmtx);
		if(malludpsockets.find(soc) == malludpsockets.end())
			return -1;

		MonUdpSocketInstance::UDPMSG* pmsg = new MonUdpSocketInstance::UDPMSG();
		memcpy(pmsg->mpbuf, buf, len);
		pmsg->mlen = len;
		pmsg->maddr = *to;
		pmsg->soc = soc;
		malludpsockets[soc].vsendbuf.push_back(pmsg);
		return len;
	}

	int SocketMulplexService::recvfrom(SOCKET soc, char *buf, int len, sockaddr *from)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malludpsockets.find(soc) == malludpsockets.end())
			return -1;

		if(malludpsockets[soc].vrecvbuf.size() == 0)
			return 0;

		int finallen = len;
		if(len < malludpsockets[soc].vrecvbuf[0]->mlen)
			finallen = malludpsockets[soc].vrecvbuf[0]->mlen;

		memcpy(buf, malludpsockets[soc].vrecvbuf[0]->mpbuf, finallen);

		//drop excess data
		delete malludpsockets[soc].vrecvbuf[0];
		malludpsockets[soc].vrecvbuf.erase(malludpsockets[soc].vrecvbuf.begin());

		return finallen;
	}

	int SocketMulplexService::limitnetspeed(int quota, int direction)	//byte quota and directory
	{
		if(direction == 0)
			max_dquota = quota;
		else
			max_uquota = quota;
        M_DEBUG("limitnetspeed Max QUota:%d\n", max_uquota);
		return 0;
	}

    int SocketMulplexService::set_recvbufsize(int bufsize)
    {
        mrecvbufsize = bufsize;
        return 0;
    }

} }
